package com.gxg.mr;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

/**
 * 需求：读取HDFS上的hello.txt文件，计算文件中每个单词出现的总次数
 *
 * 原始文件hello.txt的内容：
 * hello you
 * hello me
 *
 * 最终输出的结果形式如下：
 * hello    2
 * me   1
 * you  1
 */
public class WordCountJob {

    /**
     * 组装job=map+reduce
     */
    public static void main(String[] args) {
        try {
            if(args.length != 2){
                //如果传递的参数不够，程序直接退出
                System.out.println("参数传递错误，程序退出！");
                System.exit(100);
            }

            //执行job需要的配置参数
            Configuration conf = new Configuration();
            //创建一个job
            Job job = Job.getInstance(conf);

            //注意：必须设置该行代码，否则在集群中执行时找不到wordCountJob这个类
            job.setJarByClass(WordCountJob.class);

            //指定输入路径(可以是文件，可以是目录)
            FileInputFormat.setInputPaths(job, new Path(args[0]));

            //指定输出路径(只能指定一个不存在的目录)
            FileOutputFormat.setOutputPath(job, new Path(args[1]));

            //指定map相关代码
            job.setMapperClass(MyMapper.class);
            //指定k2类型
            job.setMapOutputKeyClass(Text.class);
            //指定v2类型
            job.setMapOutputValueClass(LongWritable.class);

            //指定reduce相关代码
            job.setReducerClass(myReducer.class);
            //指定k3类型
            job.setOutputKeyClass(Text.class);
            //指定v3类型
            job.setOutputValueClass(LongWritable.class);

            //提交job
            job.waitForCompletion(true);

        }catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * 实现Map阶段的map函数
     * 其中：Mapper<k1, v1, k2, v2>
     */
    public static class MyMapper extends Mapper<LongWritable, Text, Text, LongWritable> {

        /**
         * 实现map函数
         * 这个map函数接收<k1,v1>，产生<k2, v2>
         * @param k1  表示每一行数据的行首偏移量
         * @param v1  表示每一行内容
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        protected void map(LongWritable k1, Text v1, Context context)
                throws IOException, InterruptedException {
            //对获取到的每一行数据进行切割，把单词切割出来
            String[] words = v1.toString().split(" ");
            //迭代切割出来的单词数据
            for (String word: words) {
                //把迭代出来的单词封装成<k2, v2>的形式
                Text k2 = new Text(word);
                LongWritable v2 = new LongWritable(1L);
                //把<k2, v2>写出去
                context.write(k2, v2);
            }
        }
    }

    /**
     * 实现Reduce阶段的reduce函数
     * 其中：Reducer<k2, v2, k3, v3>
     */
    public static class myReducer extends Reducer<Text, LongWritable, Text, LongWritable>{

        /**
         * 实现reduce函数
         * 针对<k2, {v2...}的数据进行累加求和，并且最终将数据转换为<k3,v3>写出去
         * @param k2
         * @param v2s
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        protected void reduce(Text k2, Iterable<LongWritable> v2s, Context context)
                throws IOException, InterruptedException {
            //保存v2s的累加和，即v3
            long sum = 0L;
            //迭代v2s，对v2s中的数据累加求和
            for (LongWritable v2: v2s) {
                sum += v2.get();
            }

            //组装k3, v3
            Text k3 = k2;
            LongWritable v3 = new LongWritable(sum);

            //把<k3, v3>写出去
            context.write(k3, v3);
        }
    }

}
